 1.Spark Streaming之DStream基础数据源下文件数据流
   
   基础数据源包括：文件数据流、socket数据流、RDD队列流；这些数据源主要用于测试。
   引入依赖：
<dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming_2.12</artifactId>
      <version>${spark.version}</version>
</dependency>
   
   文件数据流：通过 textFileStream(directory) 方法进行读取 HDFS 兼容的文件系统
文件
   Spark Streaming 将会监控 directory 目录，并不断处理移动进来的文件
       不支持嵌套目录
       文件需要有相同的数据格式
       文件进入 directory 的方式需要通过移动或者重命名来实现
       一旦文件移动进目录，则不能再修改，即便修改了也不会读取新数据
       文件流不需要接收器（receiver），不需要单独分配CPU核
package cn.lagou.Streaming.basic

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object FileDSream {
  def main(args: Array[String]): Unit = {
    // 初始化
    Logger.getLogger("org").setLevel(Level.WARN)
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName)
      .setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(10))
    ssc.sparkContext.setLogLevel("warn")

    // 创建DStream
    val lines: DStream[String] = ssc.textFileStream("data/log")

    // DStream转换
    val words: DStream[String] = lines.flatMap(_.split("\\s+"))
    val result: DStream[(String, Int)] = words.map((_, 1)).reduceByKey(_ + _)
    // DStream输出操作
    result.print(20)

    // 启动作业
    ssc.start()
    ssc.awaitTermination()
  }
}

 
 2.Socket数据流
   
   Spark Streaming可以通过Socket端口监听并接收数据，然后进行相应处理；
   新开一个命令窗口，启动 nc 程序：
   nc -lk 9999
   # yum install nc
   随后可以在nc窗口中随意输入一些单词，监听窗口会自动获得单词数据流信息，在
监听窗口每隔x秒就会打印出词频统计信息，可以在屏幕上出现结果。
   备注：使用local[*]，可能存在问题。
   如果给虚拟机配置的cpu数为1，使用local[*]也只会启动一个线程，该线程用于
receiver task，此时没有资源处理接收达到的数据。
   【现象：程序正常执行，不会打印时间戳，屏幕上也不会有其他有效信息】
   注意：DStream的 StorageLevel 是 MEMORY_AND_DISK_SER_2；
package cn.lagou.Streaming.basic

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SocketDSream {
  def main(args: Array[String]): Unit = {
    // 初始化
    Logger.getLogger("org").setLevel(Level.WARN)
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName)
      .setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(10))
    ssc.sparkContext.setLogLevel("warn")

    // 创建DStream
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("linux122", 9999)

    // DStream转换
    val words: DStream[String] = lines.flatMap(_.split("\\s+"))
    val result: DStream[(String, Int)] = words.map((_, 1)).reduceByKey(_ + _)
    // DStream输出操作
    result.print(20)

    // 启动作业
    ssc.start()
    ssc.awaitTermination()
  }
}
   
   SocketServer程序（单线程），监听本机指定端口，与socket连接后可发送信息：
package cn.lagou.Streaming.basic

import java.io.PrintWriter
import java.net.{ServerSocket, Socket}

import scala.util.Random

object SocketServerLikeNC {
  def main(args: Array[String]): Unit = {
    val words: Array[String] =
      ("Hello World Hello Hadoop Hello spark kafka " +
        "hive zookeeper hbase flume sqoop").split("\\s+")
    val n: Int = words.length
    val port: Int = 9999
    val random: Random = scala.util.Random
    
    val server = new ServerSocket(port)
    val socket: Socket = server.accept()
    println("成功连接到本地主机：" + socket.getInetAddress)
    while (true) {
      val out = new PrintWriter(socket.getOutputStream)
      out.println(words(random.nextInt(n)) + " " + words(random.nextInt(n)))
      out.flush()
      Thread.sleep(100)
    }
  }
}
   
   SocketServer程序（多线程）
package cn.lagou.streaming
import java.net.ServerSocket

object SocketServer {
  def main(args: Array[String]): Unit = {
    val server = new ServerSocket(9999)
    println(s"Socket Server 已启动：
${server.getInetAddress}:${server.getLocalPort}") 
    while (true) {
    val socket = server.accept()
    println("成功连接到本地主机：" + socket.getInetAddress)
    new ServerThread(socket).start()
    }
  }
}  

package cn.lagou.streaming
import java.io.DataOutputStream
import java.net.Socket
class ServerThread(sock: Socket) extends Thread {
  val words = "hello world hello spark hello word hello java hello hadoop hello kafka".split("\\s+")
  val length = words.length
  
  override def run(): Unit = {
    val out = new DataOutputStream(sock.getOutputStream)
    val random = scala.util.Random
    while (true) {
      val (wordx, wordy) = (words(random.nextInt(length)),
words(random.nextInt(length)))
      out.writeUTF(s"$wordx $wordy")
      Thread.sleep(100)
    }
  }
}
 3.RDD队列流
   
   调试Spark Streaming应用程序的时候，可使用streamingContext.queueStream
(queueOfRDD)创建基于RDD队列的DStream；
   备注：
        oneAtATime：缺省为true，一次处理一个RDD；设为false，一次处理全部RDD
        RDD队列流可以使用local[1]
        涉及到同时出队和入队操作，所以要做同步
   每秒创建一个RDD（RDD存放1-100 的整数)，Streaming每隔1秒就对数据进行处
理，计算RDD中数据除10取余的个数。
package cn.lagou.Streaming.basic

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}

import scala.collection.mutable

object RDDQueueDStream {
  def main(args: Array[String]): Unit = {
    // 初始化
    Logger.getLogger("org").setLevel(Level.WARN)
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName)
      .setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(1))
    ssc.sparkContext.setLogLevel("warn")

    // 创建DStream
    val queue = new mutable.Queue[RDD[Int]]()
    val queueDStream: InputDStream[Int] = ssc.queueStream(queue)

    val result: DStream[(Int, Int)] = queueDStream.map(elem => (elem % 10, 1))
      .reduceByKey(_ + _)
    result.print()

    // 启动作业
    ssc.start()
    // 每秒生成一个RDD,将RDD放置在队列中
    for (i <- 1 to 5) {
      queue.synchronized{
        val range = (1 to 100).map(_*i)
        queue += ssc.sparkContext.makeRDD(range, 2)
      }
      Thread.sleep(1000)
    }
    ssc.stop()
  }
}
   